package com.huan.network;

import com.huan.bean.ApacheLogEvent;
import com.huan.bean.PageViewCount;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.net.URL;
import java.text.SimpleDateFormat;
import java.util.regex.Pattern;


//常规类
public class HotPages {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic( TimeCharacteristic.EventTime );
        env.setParallelism( 1 );

        // 读取文件，转换成POJO
       URL resource = HotPages.class.getResource("/apache.log");
       DataStream<String> inputStream = env.readTextFile(resource.getPath());
        //DataStream<String> inputStream = env.socketTextStream( "localhost", 7777 );

        DataStream<ApacheLogEvent> dataStream = inputStream
                .map( line -> {
                    String[] fields = line.split( " " );
                    //定义时间，转化时间格式
                    SimpleDateFormat simpleDateFormat = new SimpleDateFormat( "dd/MM/yyyy:HH:mm:ss" );
                    //解析时间并且获取
                    Long timestamp = simpleDateFormat.parse( fields[3] ).getTime();
                    return new ApacheLogEvent( fields[0], fields[1], timestamp, fields[5], fields[6] );
                } )
                .assignTimestampsAndWatermarks( new BoundedOutOfOrdernessTimestampExtractor<ApacheLogEvent>( Time.seconds( 1 ) ) {
                    @Override
                    public long extractTimestamp(ApacheLogEvent element) {
                        return element.getTimestamp();
                    }
                } );

       //dataStream.print( "data" );

        // 分组开窗聚合
        SingleOutputStreamOperator<PageViewCount> windowAggStream = dataStream
                .filter( data -> "GET".equals( data.getMethod() ) )    // 过滤get请求
                .filter(data -> {
                    String regex = "^((?!\\.(css|js|png|ico)$).)*$";
                    return Pattern.matches(regex, data.getUrl());
                })
                .keyBy( ApacheLogEvent::getUrl )    // 按照url分组
                .timeWindow( Time.minutes( 10 ), Time.seconds( 5 ) )
                .aggregate( new PageCountAgg(), new PageCountResult() );


        // 收集同一窗口count数据，排序输出
        DataStream<String> resultStream = windowAggStream
                .keyBy( PageViewCount::getWindowEnd )
                .process( new TopNHotPages( 3 ) );

        resultStream.print();

        env.execute( "hot pages job" );
    }
}
